unit PtActions;

interface

//  Actions must implement the interface IAction.  The class TAction is provided
//  in case you don't want to explicitly use an interface.  In effect, it makes
//  the use of IAction optional (including interface referece counting, etc).

//  There are two main types of action handlers...
//  1) Parallel processing
//  This is used when you want to run many threads simultaneously.
//  - This action handler is called TActionHandler.
//  - The number of threads it runs is specified by the ThreadCount property.
//  - Errors are propagated via the event OnError.
//  - Actions can specify their thread priority by also implementing the
//    interface IActionPriority.
//
//  2) Serial processing
//  This is useful for synchronising actions.  This can be used as an alternative
//  to TThread.Synchronise() (which requires the Windows event queue).
//  For example, you might want a thread dedicated to ActiveX.  The Initialise
//  and Finalise methods would then contain CoInitialize() and CoUninitialize.
//  This would then allow the rest of the application to be multithreaded.
//  - The methods Initialise and Finalise are called by the thread which executes
//    the actions.
//  - In the constructor you specify whether you wish to use the main thread or
//    a dedicated thread.
//  - Only available in Windows OS.

uses
  Classes, SysUtils, SyncObjs, PtADTIntf;

type
  //  All actions must implement IAction

  IAction = interface
  ['{55FDF085-2D9E-406E-BE6E-19C4A0616DF6}']
    procedure Execute;
  end;

  //  TAction is available so that an action can be used as a plain object.
  //  NOTE: TAction does *not* use reference counting.

  TAction = class(TObject, IInterface, IAction)
  private
    function QueryInterface(const IID: TGUID; out Obj): HResult; stdcall;
    function _AddRef: Integer; stdcall;
    function _Release: Integer; stdcall;
  public
    procedure Execute; virtual; abstract;
  end;

  //  Parallel action handler

  //  Actions may also implement IActionPriority

  IActionPriority = interface
  ['{CEA25500-BB27-46A8-B675-B4B2D2630136}']
{$IFDEF MSWINDOWS}
    function GetPriority: TThreadPriority;
    property Priority : TThreadPriority read GetPriority;
{$ENDIF}
{$IFDEF LINUX}
    function GetPriority: integer;
    function GetPolicy: integer;
    property Priority : Integer read GetPriority;
    property Policy   : Integer read GetPolicy;
{$ENDIF}
  end;

  TActionError = procedure (const Action: IAction; const E: Exception) of object;

  TActionHandler = class;

  TActionThread = class(TThread)
  private
    fOwner: TActionHandler;
  protected
    procedure SetDefaultPriority;
    procedure Execute; override;
    procedure ExecuteAction(Action: IAction); virtual;
  public
    constructor Create(const AOwner: TActionHandler);
    property  Owner : TActionHandler  read fOwner;
  end;
  TActionThreadClass = class of TActionThread;

  TActionHandler = class
  private
    fThreads          : TList;
    fCriticalSection  : TCriticalSection;
    fOnError          : TActionError;
    fActions          : TQueue;
{$IFDEF MSWINDOWS}
    fDefaultPriority  : TThreadPriority;
{$ENDIF}
{$IFDEF LINUX}
    fDefaultPriority  : Integer;
    fDefaultPolicy    : Integer;
{$ENDIF}
    function  GetThreadCount: integer;
    procedure SetThreadCount(const Value: integer);
  protected
    property  CriticalSection : TCriticalSection  read fCriticalSection;
    function  Pop: IAction;
    procedure WakeAThread;
    procedure DoError(const Action: IAction; const E: Exception);
    function  CreateThread: TActionThread; virtual;
  public
    constructor Create(const AThreadCount: integer = 0);
    destructor  Destroy; override;
    procedure Add(Action: IAction);
    property  ThreadCount : integer       read GetThreadCount write SetThreadCount;
    property  OnError     : TActionError  read fOnError       write fOnError;
{$IFDEF MSWINDOWS}
    property DefaultPriority : TThreadPriority read fDefaultPriority  write fDefaultPriority;
{$ENDIF}
{$IFDEF LINUX}
    property DefaultPriority : Integer read fDefaultPriority  write fDefaultPriority;
    property DefaultPolicy   : Integer read fDefaultPolicy    write fDefaultPolicy;
{$ENDIF}
  end;

  EActionHandler = class(Exception)
  end;

implementation

{ TAction }

function TAction._AddRef: Integer;
begin
  Result  := 0
end;

function TAction._Release: Integer;
begin
  Result  := 0
end;

function TAction.QueryInterface(const IID: TGUID; out Obj): HResult;
begin
  if GetInterface(IID, Obj) then
    Result := 0
  else
    Result := E_NOINTERFACE;
end;

{ TActionThread }

constructor TActionThread.Create(const AOwner: TActionHandler);
begin
  inherited Create( false );
  fOwner  := AOwner;
end;

procedure TActionThread.Execute;
var
  Action    : IAction;
  Priority  : IActionPriority;
begin
  while not Terminated  do
    begin
      try
        Action  := Owner.Pop;
        if  Action = nil  then
          //  No work to do, so go to sleep
          Suspend
        else
          begin
            //  Give the thread desired priority
            if  Supports(Action, IActionPriority, Priority) then
              begin
                Self.Priority := Priority.Priority;
{$IFDEF LINUX}
                Self.Policy   := Priority.Policy;
{$ENDIF}
              end
            else
              SetDefaultPriority;
            //  Execute the action
            ExecuteAction( Action );
          end;
      //  Might have to capture exception
      except
        on  E: Exception  do
          Owner.DoError( Action, E );
      end;
      //  Reset thread priority
      SetDefaultPriority;
    end;
end;

procedure TActionThread.ExecuteAction(Action: IAction);
begin
  Action.Execute;
end;

procedure TActionThread.SetDefaultPriority;
begin
  Self.Priority := Owner.DefaultPriority;
{$IFDEF LINUX}
  Self.Policy   := Owner.DefaultPolicy;
{$ENDIF}
end;

{ TActionHandler }

procedure TActionHandler.Add(Action: IAction);
begin
  CriticalSection.Enter;
  try
    fActions.Push( Action );
  finally
    CriticalSection.Leave;
  end;
  //  If a thread is suspended, then wake it up
  WakeAThread;
end;

constructor TActionHandler.Create(const AThreadCount: integer);
begin
  inherited Create;
  fCriticalSection  := TCriticalSection.Create;
  fThreads          := TList.Create;
  fActions          := TQueue.Create;
{$IFDEF MSWINDOWS}
  fDefaultPriority  := tpNormal;
{$ENDIF}
{$IFDEF LINUX}
  fDefaultPriority  := Insert a good default value;
  fDefaultPolicy    := Insert a good default value;
{$ENDIF}
  ThreadCount       := AThreadCount;
end;

function TActionHandler.CreateThread: TActionThread;
begin
  Result  := TActionThread.Create(Self);
end;

destructor TActionHandler.Destroy;
begin
  ThreadCount       := 0;
  fThreads.Free;
  fActions.Free;
  fCriticalSection.Free;
  inherited;
end;

procedure TActionHandler.DoError(const Action: IAction; const E: Exception);
begin
  if  Assigned(fOnError)  then
    fOnError( Action, E )
end;

function TActionHandler.GetThreadCount: integer;
begin
  Result  := fThreads.Count;
end;

function TActionHandler.Pop: IAction;
begin
  CriticalSection.Enter;
  try
    if  not fActions.IsEmpty  then
      Supports( fActions.Pop, IAction, Result )
    else
      Result  := nil;
  finally
    CriticalSection.Leave;
  end;
end;

procedure TActionHandler.SetThreadCount(const Value: integer);
var
  i : integer;
  Thread  : TActionThread;
begin
  if  Value < 0 then
    raise EActionHandler.Create(ClassName+' cannot have a negative number of threads');
  CriticalSection.Enter;
  try
    //  May want to increase the number of threads
    if  Value > fThreads.Count  then
      for i := fThreads.Count+1 to Value  do
        fThreads.Add( CreateThread );
    //  May want to reduce the number of threads
    if  Value < fThreads.Count  then
      begin
        for i := fThreads.Count-1 downto Value  do
          begin
{$WARN UNSAFE_CAST OFF}
            Thread  := TActionThread( fThreads[i] );
{$WARN UNSAFE_CAST ON}
            Thread.FreeOnTerminate  := True;
            Thread.Terminate;
            Thread.Resume;  //  Just in case the thread is sleeping
          end;
        fThreads.Count  := Value;
      end;
  finally
    CriticalSection.Leave;
  end;
end;

procedure TActionHandler.WakeAThread;
var
  i : integer;
begin
  CriticalSection.Enter;
  try
    for i := 0 to fThreads.Count-1  do
{$WARN UNSAFE_CAST OFF}
      if  TActionThread(fThreads[i]).Suspended  then
        begin
          TActionThread(fThreads[i]).Resume;
          break;
        end;
{$WARN UNSAFE_CAST ON}
  finally
    CriticalSection.Leave;
  end;
end;

end.
